package com.niit.sparkAnalyze.sparkStreaming

object ValidNum {
  //订单是否有效的总数 需求（2
  def processOrders(rdd: org.apache.spark.rdd.RDD[(String, String, Int, String, String)]): String = {
    if (!rdd.isEmpty()) {
      val validOrders = rdd.filter(_._5 == "Y").map(record => record._3.toLong)
      val invalidOrders = rdd.filter(_._5 == "N").map(record => record._3.toLong)

      val validOrdersCount = if (validOrders.isEmpty()) 0L else validOrders.reduce(_ + _)
      val invalidOrdersCount = if (invalidOrders.isEmpty()) 0L else invalidOrders.reduce(_ + _)


      //test
//      println(s"有效订单总数: $validOrdersCount")
//      println(s"无效订单总数: $invalidOrdersCount")

      println("(sparkStreaming)"+"有效订单总数:"+validOrdersCount+", 无效订单总数:"+invalidOrdersCount)
      "有效订单总数:"+validOrdersCount+", 无效订单总数:"+invalidOrdersCount
    } else {
      "RDD is empty"
    }
  }
}
